package yunjiao.javatutorials.apache.kafka;

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.TopicPartitionInfo;

import java.util.*;
import java.util.concurrent.ExecutionException;

/**
 * 监控和工具类
 *
 * @author yangyunjiao
 */
public class KafkaMonitor {

    private final AdminClient adminClient;

    public KafkaMonitor(String bootstrapServers) {
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        this.adminClient = AdminClient.create(props);
    }

    /**
     * 获取主题列表
     */
    public Set<String> listTopics() throws ExecutionException, InterruptedException {
        ListTopicsResult result = adminClient.listTopics();
        return result.names().get();
    }

    /**
     * 获取主题详情
     */
    public Map<String, TopicDescription> describeTopics(Collection<String> topics)
            throws ExecutionException, InterruptedException {
        DescribeTopicsResult result = adminClient.describeTopics(topics);
        return result.all().get();
    }

    /**
     * 获取消费者组列表
     */
    public Collection<ConsumerGroupListing> listConsumerGroups()
            throws ExecutionException, InterruptedException {
        ListConsumerGroupsResult result = adminClient.listConsumerGroups();
        return result.all().get();
    }

    /**
     * 检查消费者延迟
     */
    public Map<TopicPartition, Long> getConsumerLag(String groupId, String topic)
            throws ExecutionException, InterruptedException {
        Map<TopicPartition, Long> lags = new HashMap<>();

        // 获取消费者组的偏移量
        DescribeConsumerGroupsResult groupResult = adminClient.describeConsumerGroups(Collections.singleton(groupId));
        ConsumerGroupDescription groupDescription = groupResult.all().get().get(groupId);

        // 获取主题的末端偏移量
        ListTopicsResult topicsResult = adminClient.listTopics();
        Set<String> topics = topicsResult.names().get();

        if (topics.contains(topic)) {
            DescribeTopicsResult topicsDescResult = adminClient.describeTopics(Collections.singleton(topic));
            TopicDescription topicDescription = topicsDescResult.all().get().get(topic);

            for (TopicPartitionInfo partitionInfo : topicDescription.partitions()) {
                TopicPartition topicPartition = new TopicPartition(topic, partitionInfo.partition());

                // 这里简化实现，实际应该获取消费者组的当前偏移量并与日志末端偏移量比较
                // 实际生产环境建议使用专门的监控工具
            }
        }

        return lags;
    }

    public void close() {
        if (adminClient != null) {
            adminClient.close();
        }
    }

    public static void main(String[] args) throws Exception {
        KafkaMonitor monitor = new KafkaMonitor("localhost:9092");

        try {
            // 列出所有主题
            Set<String> topics = monitor.listTopics();
            System.out.println("Kafka主题列表: " + topics);

            // 获取主题详情
            if (!topics.isEmpty()) {
                Map<String, TopicDescription> topicDetails = monitor.describeTopics(topics);
                topicDetails.forEach((topicName, description) -> {
                    System.out.println("主题: " + topicName + ", 分区数: " + description.partitions().size());
                });
            }

        } finally {
            monitor.close();
        }
    }
}
